import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Counters;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.jobcontrol.Job;

public class Grafo {

	public static class PreProcessingMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		/*
		 * Esse mapper é o responsável pelo preprocessamento da base de dados,
		 * ele recebe como entrada a base de dados e uma lista de usuários
		 * iniciais A chave de entrada será o número da linha e o valor a linha.
		 */

		private Text node = new Text();
		private Text word = new Text();

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {
			String line = value.toString();
			// Aqui verificamos se a linha de entrada pertence a base de dados
			// ou a lista de usuários
			if (line.split(",").length == 4) {

				StringTokenizer tokenizer = new StringTokenizer(line, ",");

				String node = tokenizer.nextToken();
				// Setamos como node o userID do usuário que adicionou outro
				// como contato
				this.node.set(node.trim());
				// Aqui setamos o usuário que foi marcado como contato
				word.set(tokenizer.nextToken().trim());

				// Verificamos se foi um evento de adicionar contato
				if (tokenizer.nextToken().trim().equalsIgnoreCase("0")) {
					// Enviavos para o reducer o usuário que adicionou e o que
					// foi adicionado como contato
					output.collect(this.node, word);
				}
			} else {
				// Se a entrade vier da lista de usuários
				if (!line.equals("")) {
					this.node.set(line);
					// Emitimos para o reducer o userID e a flag INIT dizendo
					// que esse ID tá na lista inicial
					output.collect(node, new Text("INIT"));
				}
			}
		}
	}

	public static class PreProcessingReduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {

		/*
		 * Esse reducer é responsável por pegar os dados do mapper, chave
		 * userID, valor userID ou INIT e montar um grafo de amizade no formato
		 * userID \t 1 or 0, userID userID userID ..., vai ser 1 se ele estiver
		 * na lista de usuários, 0 caso contrário e os userIDs após a vírgula
		 * são os amigos dele
		 */

		private Text node = new Text();
		private Text word = new Text();
		public static String rootNodes;

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {

			boolean inicial = false;
			String amigos = "";

			// Iteração na lista de amigos
			while (values.hasNext()) {
				Text value = values.next();
				// se encontrou INIT, pertence a lista inicial
				if (value.equals(new Text("INIT"))) {
					inicial = true;
				} else {
					// Se não for init adicionar o userID para a lista de amigos
					amigos += value.toString().trim() + " ";
				}
			}

			// String que vai ficar no novo grafo, indicando que o usuário é ou
			// não nó inicial
			String str = "0";

			if (inicial) {
				// se for inicial colocamos o valor 1
				str = "1";
			}

			str += "," + amigos;
			str = str.substring(0, str.length() - 1);
			word.set(str);
			// Emitimos o userID ,1 se ele estava na lista inicial e 0 caso
			// contrário e a lista de amigos
			output.collect(key, word);
		}

	}

	public static class Map extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {

		/*
		 * Esse mapper pega como chave o userID do grafo e como valor
		 * 1ou0,userID..., recebe também como entrada a lista de usuários E vai
		 * ficar responsável por pegar os amigos dos usuários que tem valor 1 e
		 * passar adiante
		 */

		private Text node = new Text();
		private Text word = new Text();

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {
			String line = value.toString();

			// Verifica se pertence ao Grafo ou a lista de usuário
			if (line.split("\t").length == 2) {
				StringTokenizer tokenizer = new StringTokenizer(line, ",");

				String node = tokenizer.nextToken();
				// setamos o userID do usuário
				String user = node.split("\t")[0];
				this.node.set(user);
				// iterando na lista de amigo
				while (tokenizer.hasMoreElements()) {
					word.set(tokenizer.nextToken().trim());
					// emitimos o usuário e seus amigos
					output.collect(this.node, word);
				}

			} else {
				if (!line.equals("")) {
					this.node.set(line);
					// emitimos o userID e INIT caso ele pertenca a lista
					// inicial
					output.collect(node, new Text("INIT"));
				}
			}
		}
	}

	public static class Reduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {

		/*
		 * Esse reducer vai ficar responsável por receber o grafo e emitir o
		 * grafo atualizado
		 */

		private Text node = new Text();
		private Text word = new Text();
		public static String rootNodes;

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {

			boolean inicial = false;
			String amigos = "";
			// Iteração na lista de amigos
			while (values.hasNext()) {
				Text value = values.next();
				// Se tem o valor INIT, então estava na lista de usuários
				if (value.equals(new Text("INIT"))) {
					inicial = true;
				} else {
					// O que não é INIT é adicionado na lista de amigos do
					// usuário
					amigos += value.toString().trim() + " ";
				}
			}

			// String que vai ficar no novo grafo, indicando que o usuário é ou
			// não nó inicial
			String str = "0";

			if (inicial) {
				// se for inicial colocamos o valor 1
				str = "1";
			}

			str += "," + amigos;
			str = str.substring(0, str.length() - 1);
			word.set(str);
			// Emitimos o userID ,1 se ele estava na lista inicial e 0 caso
			// contrário e a lista de amigos
			output.collect(key, word);
		}

	}

	public static class Map2 extends MapReduceBase implements
			Mapper<Text, Text, Text, Text> {

		/*
		 * Esse mapper pega o grafo como entrada, chave é o userID e valor
		 * 1ou0,userID... e emite para o reducer todos os amigos do user que tem
		 * como valor 1
		 */
		private Text node = new Text();
		private Text word = new Text();

		@Override
		public void map(Text key, Text value,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {

			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line, ",");

			// Pegando a flag 1 ou 0
			String flag = tokenizer.nextToken().trim();
			// Verifica se o usuário tem amigos
			if (tokenizer.hasMoreElements()) {
				// Pegando a lista de amigos do usários
				StringTokenizer vizinhos = new StringTokenizer(tokenizer
						.nextToken().trim());

				word.set("1");
				// se tinha valor 1, ou seja, era da lista de usuários
				if (flag.equalsIgnoreCase("1")) {
					// Esse output coloca o usuário inicial na saida
					output.collect(key, value);
					while (vizinhos.hasMoreTokens()) {
						// Adicionamos todos os amigos do usuário na saida
						node.set(vizinhos.nextToken().trim());
						output.collect(node, word);
					}
				}
			} else {
				// Se não tem amigos emite apenas o usuário
				if (flag.equalsIgnoreCase("1")) {
					output.collect(key, value);
				}
			}
		}

	}

	public static class Reduce2 extends MapReduceBase implements
			Reducer<Text, Text, Text, NullWritable> {

		/*
		 * Esse reducer recebe os usuários e emite na saida a lista dos usuários
		 * e incrementa o contador de usuários adicionados
		 */

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, NullWritable> output, Reporter arg3)
				throws IOException {

			// Incremento do contador
			arg3.incrCounter("Usuarios", "processados", 1);
			// Emitimos o usuário
			output.collect(key, NullWritable.get());
		}

	}

	public static void main(String[] args) throws IOException {
		// Criacao do contador, RunningJob para pegar informação
		RunningJob rjob;
		Counters myCounters;
		Counter contador;
		//Criação de configuração para remoção de pastas
		Configuration removeConf = new Configuration();
		removeConf.set("fs.default.name","hdfs://127.0.0.1:9000/");
		FileSystem hdfs = FileSystem.get(removeConf);
		
		// Configuração do job
		JobConf preprocessing = new JobConf(Grafo.class);
		preprocessing.setJobName("PreProcessing");
		preprocessing.setOutputKeyClass(Text.class);
		preprocessing.setOutputValueClass(Text.class);

		preprocessing.setMapperClass(PreProcessingMap.class);
		preprocessing.setReducerClass(PreProcessingReduce.class);

		preprocessing.setInputFormat(TextInputFormat.class);
		preprocessing.setOutputFormat(TextOutputFormat.class);

		// Passando dois arquivos de entrada
		FileInputFormat.addInputPaths(preprocessing, "./base,./list_1");
		FileOutputFormat.setOutputPath(preprocessing, new Path("./tmp_1"));

		JobClient.runJob(preprocessing);

		/*
		 * Iteração, cada iteração vai ser pego o grafo e a lista de usuários, e
		 * vai ser atualizado a lista de usuário até atingir a condição de
		 * parada
		 */
		int i = 1;
		do {
			// Configuração do job
			JobConf conf = new JobConf(Grafo.class);

			conf.setJobName("Grafo");
			conf.setOutputKeyClass(Text.class);
			conf.setOutputValueClass(Text.class);

			conf.setMapperClass(Map.class);
			conf.setReducerClass(Reduce.class);

			conf.setInputFormat(TextInputFormat.class);
			conf.setOutputFormat(TextOutputFormat.class);

			// Setando dois arquivos de entrada, que são as saídas do reduce
			// anterior
			FileInputFormat.addInputPaths(conf, "./tmp_" + i + ",./list_" + i);
			
			i++;
			// Setando o output para ./tmp_i
			FileOutputFormat.setOutputPath(conf, new Path("./tmp_" + i));

			JobClient.runJob(conf);
			
			//Remoção de pastas
			Path dir1 = new Path("./tmp_" + (i -1));
			Path dir2 = new Path("./list_" + (i-1));

			try {
				hdfs.delete(dir1);
				hdfs.delete(dir2);
			} catch (IOException e) {
				e.printStackTrace();
			}



			// Configuração do JOB
			JobConf conf2 = new JobConf(Grafo.class);

			conf2.setJobName("Grafo2");
			conf2.setOutputKeyClass(Text.class);
			conf2.setMapOutputValueClass(Text.class);
			conf2.setOutputValueClass(NullWritable.class);

			conf2.setMapperClass(Map2.class);
			conf2.setReducerClass(Reduce2.class);

			conf2.setInputFormat(KeyValueTextInputFormat.class);
			conf2.setOutputFormat(TextOutputFormat.class);

			// Setando input path para tmp_i
			FileInputFormat.setInputPaths(conf2, new Path("./tmp_" + i));
			
			// Setando output para args[0]+_+i
			FileOutputFormat.setOutputPath(conf2,
					new Path("./list" + "_" + (i)));

			// Adicionando o job1 como dependência
			Job job1 = new Job(conf);
			Job job2 = new Job(conf2);
			job2.addDependingJob(job1);

			rjob = job2.getJobClient().runJob(conf2);
			
			// Pegando o contador
			myCounters = rjob.getCounters();
			contador = myCounters.findCounter("Usuarios", "processados");
			System.out.println(contador.getValue());
			// Verificando se já chegou em 80% do total de usuários
		} while (contador.getCounter() < 73882.4);

	}
}
